package com.lwc;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.MessageProperties;

import java.io.IOException;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.TimeoutException;

//生产者  向mq 投递消息
public class Productor {
    public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
//1创建连接工厂  设置连接信息
        ConnectionFactory factory=new ConnectionFactory();
factory.setHost("192.168.81.128");//连接主机
factory.setUsername("root");//默认是guest
 factory.setPassword("root");//默认是guest
factory.setPort(5672);//连接端口号
factory.setVirtualHost("erp");//虚拟主机

        //2用连接工厂创建连接
        Connection conn=factory.newConnection();

        //3创建信道
        Channel channel=conn.createChannel();

        //声明队列名称
        String queueName="confirm-async";
/**
 * queue 队列名称
 * durable 是否持久化
 *  exclusive是否独占队列   为true 信道必须是同一连接创建的
 *autodelete  是否自动删除    前提队列中灭有消息
 *arguments 向队列传递的其他参数
 *
 */
        //声明队列
        channel.queueDeclare(queueName,false,false,false,null);
        //跳表   用于高并发  适用于map
        ConcurrentNavigableMap<Long, String> skipListMap = new ConcurrentSkipListMap<>();
        ConcurrentNavigableMap<Long, String> failListMap = new ConcurrentSkipListMap<>();
//5.5增加监听   等待服务器确认或返回

        channel.addConfirmListener((deliveryTag,multiple)->{
            //
if (multiple){
//为true  表示 所有的消息的序号小于deliveryTag 的消息 如 2 6 5 4 8 如果返回8   代表全部得到了服务器确认
//需要将确认的移除
    //返回key值小于等于deliveryTag的集合   子集（不需要遍历）
    ConcurrentNavigableMap childMap=skipListMap.headMap(deliveryTag);
//将子集从调表中移除
childMap.clear();//将子集中所有的数据清楚
 }else {
    //如果为false表示单个移除
    skipListMap.remove(deliveryTag);
}
        },(deliveryTag,multiple)->{
            //消息被服务器否定后的回调    没有发布成功

            if (multiple){
                //返回key值小于等于deliveryTag的集合   子集（不需要遍历）
                ConcurrentNavigableMap childMap=skipListMap.headMap(deliveryTag);
              failListMap.putAll(childMap);
            }else {
//将失败的消息记录
                failListMap.put(deliveryTag,skipListMap.get(deliveryTag));
            }
        });


        long begin=System.currentTimeMillis();
//6.消息发布
        for (int i = 0; i <1000 ; i++) {
            String msg="消息"+i;
            //获取消息发布的序列号
            Long seq=channel.getNextPublishSeqNo();

//将消息与序列号关联起来
            skipListMap.put(seq,msg);
            //发布消息
            channel.basicPublish("",queueName,MessageProperties.PERSISTENT_TEXT_PLAIN,msg.getBytes());
        }


        System.out.println("耗费时间"+(System.currentTimeMillis()-begin));



        //如果调表还有数据
        if (skipListMap.size()>0){
            System.in.read();
        }


        //释放资源
        channel.close();
        conn.close();

    }
}
